package cn.qdhlm.testrocketmq.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @Name:Consumer2
 * @description:consumer2
 * @author: xinY
 **/
public class Consumer2 {
    public static void main(String[] args) throws Exception{
        // Instantiate with specified consumer group name.
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer");
        // Specify name server addresses.
//        consumer.setNamesrvAddr("localhost:9876");
        consumer.setNamesrvAddr("192.168.0.159:9876");
        consumer.setVipChannelEnabled(false);
//        consumer.setNamesrvAddr("192.168.255.132:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // Subscribe one more more topics to consume.
        consumer.subscribe("TopicTest", "*");

        // 设置消息模式广播
//        consumer.setMessageModel(MessageModel.BROADCASTING);
        // 设置消息模式集群
        consumer.setMessageModel(MessageModel.CLUSTERING);
        // Register callback to execute on arrival of messages fetched from brokers.
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.print(Thread.currentThread().getName() + "收到消息:");
                    System.out.println("topic:"+msg.getTopic()+", tags:"+msg.getTags()+", body:"+new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        //Launch the consumer instance.
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}
